package com.gukun.springboot.tdengine.service.select;

import cn.hutool.core.date.DateUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.gukun.springboot.tdengine.dao.SelectTaskMapper;
import com.gukun.springboot.tdengine.domain.exception.util.ServiceExceptionUtil;
import com.gukun.springboot.tdengine.domain.select.SelectTask;
import com.gukun.springboot.tdengine.service.backuprecover.config.ThreadPool;
import com.taosdata.jdbc.TSDBDriver;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author: 4998
 * @createDate: 2025-06-11  13:49
 * @description:
 */
@Service
@Slf4j
public class SelectTaskService {
    @Autowired
    private SelectTaskMapper selectTaskMapper;

    private static String SELECT_LAST = "select last_row(*),tbname from tfloat where tbname in (%s) partition by tbname\n" +
            "union all \n" +
            "select last_row(*),tbname from tbool where tbname in (%s) partition by tbname\n" +
            "union all\n" +
            "select last_row(*),tbname from tdouble where tbname in (%s) partition by tbname\n" +
            "union all\n" +
            "select last_row(*),tbname from tint where tbname in (%s) partition by tbname;";
    private static String SELECT_HISTORY_INTERVAL = "select ts,first(v),tbname from %s where ts > %s and ts < %s interval(%s) FILL(NEXT)";
    private static String SELECT_HISTORY = "select ts,v,tbname from %s where ts > %s and ts < %s";
    private static Map<String, CompletableFuture<?>> RUNNING_TASKS = new ConcurrentHashMap<>();

    private static ThreadPoolExecutor THREAD_POOL = ThreadPool.threadPoolExecutor(600,10,1000);
    public boolean saveOrUpdate(SelectTask selectTask) throws Exception{
        String id = selectTask.getId();
        if(StringUtils.isBlank(id)){
            selectTask.setCreateTime(DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss"));
            return selectTaskMapper.insert(selectTask) > 0;
        }else{
            return selectTaskMapper.updateById(selectTask) > 0;
        }
    }

    /**
     * 启动同一个组的任务
     */
    public boolean startByGroupId(int groupId){
        LambdaQueryWrapper<SelectTask> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(SelectTask::getGroupId, groupId).eq(SelectTask::getStatus, 0);
        List<SelectTask> selectTasks = selectTaskMapper.selectList(queryWrapper);
        //异步多线程启动任务
        for(SelectTask selectTask : selectTasks){
            start(selectTask);
        }
        return true;
    }

    /**
     * 启动全部命令
     */
    public boolean startAll(){
        LambdaQueryWrapper<SelectTask> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(SelectTask::getStatus, 0);
        List<SelectTask> selectTasks = selectTaskMapper.selectList(queryWrapper);
        for(SelectTask selectTask : selectTasks){
            start(selectTask);
        }
        return true;
    }

    public boolean start(SelectTask selectTask){
        selectTask.setStatus(1);
        selectTask.setLastTime(DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss"));
        selectTaskMapper.updateById(selectTask);
        CompletableFuture.runAsync(() -> {
            //可能会同时并发多个线程
            List<CompletableFuture<Map<String,Long>>> futures = new ArrayList<>();
            for(int i=0;i<selectTask.getConcurrentNum();i++){
                CompletableFuture<Map<String,Long>> future = CompletableFuture.supplyAsync(() -> {
                    try {
                        return executeTask(selectTask);
                    } catch (Exception e) {
                        selectTask.setStatus(0);
                        selectTask.setLastLog("任务异常终止："+e.getMessage());
                        selectTaskMapper.updateById(selectTask);
                        log.error("启动任务失败：",e);
                    }
                    return null;
                },THREAD_POOL);
                futures.add(future);
            }
            //等待所有任务完成
            CompletableFuture<?> all = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
            RUNNING_TASKS.put(selectTask.getId(), all);
            //获取所有任务的结果
            long totalTime = 0;
            long totalNum = 0;
            try{
                all.get();
                CompletableFuture<Map<String,Long>> temp = null;
                Map<String,Long> result = null;
                boolean flag = true;
                for (CompletableFuture<Map<String, Long>> future : futures) {
                    temp = future;
                    result = temp.get();
                    if(result == null){
                        //说明有任务没有执行成功
                        flag = false;
                        continue;
                    }
                    totalTime += result.get("totalTime");
                    totalNum += result.get("totalNum");
                }
                if(flag){
                    selectTask.setLastLog("任务执行完成，总耗时：" + totalTime + "ms, 总记录数：" + totalNum);
                }
            }catch (Exception e){
                log.error("任务执行异常", e);
                all.cancel(true);
                selectTask.setLastLog("任务异常终止："+e.getMessage());
            }
            selectTask.setStatus(0);
            selectTaskMapper.updateById(selectTask);
            RUNNING_TASKS.remove(selectTask.getId());
        });
        return true;
    }

    public boolean start(String id){
        SelectTask selectTask = selectTaskMapper.selectById(id);
        if(selectTask == null){
            throw ServiceExceptionUtil.exception0(500, "任务不存在");
        }
        if(selectTask.getStatus() == 1){
            throw ServiceExceptionUtil.exception0(500, "任务正在运行");
        }
        return start(selectTask);
    }

    public boolean stop(String id){
        CompletableFuture<?> future = RUNNING_TASKS.get(id);
        if(future != null){
            future.cancel(true);
            RUNNING_TASKS.remove(id);
        }
        LambdaQueryWrapper<SelectTask> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(SelectTask::getId, id);
        SelectTask selectTask = selectTaskMapper.selectById(id);
        selectTask.setStatus(0);
        selectTaskMapper.updateById(selectTask);
        return true;
    }

    public boolean stopByGroupId(int groupId){
        LambdaQueryWrapper<SelectTask> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(SelectTask::getGroupId, groupId).eq(SelectTask::getStatus, 1);
        List<SelectTask> selectTasks = selectTaskMapper.selectList(queryWrapper);
        for(SelectTask selectTask : selectTasks){
            CompletableFuture<?> future = RUNNING_TASKS.get(selectTask.getId());
            if(future != null){
                future.cancel(true);
                RUNNING_TASKS.remove(selectTask.getId());
            }
            selectTask.setStatus(0);
            selectTaskMapper.updateById(selectTask);
        }
        return true;
    }

    public boolean stopAll(){
        Set<String> ids = RUNNING_TASKS.keySet();
        for(String id : ids){
            CompletableFuture<?> future = RUNNING_TASKS.get(id);
            if(future != null){
                future.cancel(true);
                RUNNING_TASKS.remove(id);
            }
        }
        LambdaQueryWrapper<SelectTask> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.in(SelectTask::getId,  ids);
        List<SelectTask> selectTasks = selectTaskMapper.selectList(queryWrapper);
        for(SelectTask selectTask : selectTasks){
            selectTask.setStatus(0);
            selectTaskMapper.updateById(selectTask);
        }
        return true;
    }

    public Map<String,Long> executeTask(SelectTask selectTask) throws Exception{
        if(selectTask.getType() == 1){
            return executeRealValue(selectTask);
        }else{
            return executeHistoryValue(selectTask);
        }
    }

    public Connection getConnection(SelectTask selectTask) throws Exception {
        String jdbcUrl = String.format("jdbc:TAOS-WS://%s:%s?user=%s&password=%s", selectTask.getHost(), selectTask.getPort(), selectTask.getUser(), selectTask.getPassword());
        Class.forName("com.taosdata.jdbc.ws.WebSocketDriver");
        return DriverManager.getConnection(jdbcUrl);
    }

    public Map<String,Long> executeRealValue(SelectTask selectTask) throws Exception{
        long totalTime = 0;
        long totalNum = 0;
        try(Connection connection = getConnection(selectTask);
            Statement statement = connection.createStatement()){
            statement.execute("USE "+selectTask.getDbName());
            String prefix = selectTask.getPrefix();
            long startIndex = selectTask.getStartIndex();
            long endIndex = selectTask.getTagNum() + startIndex;
            int maxSqlLength = 100000;
            StringBuilder tableNameStr = new StringBuilder();
            String sql = null;
            for(long i=startIndex;i<endIndex;i++){
                tableNameStr.append("'").append(prefix).append(i).append("'");
                if(tableNameStr.length() > maxSqlLength || i == endIndex-1){
                    sql = String.format(SELECT_LAST, tableNameStr,tableNameStr,tableNameStr,tableNameStr);
                    long start = 0L;
                    long end = 0L;
                    ResultSet resultSet = null;
                    try{
                        synchronized(this){
                            start = System.currentTimeMillis();
                            resultSet = statement.executeQuery(sql);
                            end = System.currentTimeMillis();
                        }
                        totalTime = totalTime + (end - start);
                        while(resultSet.next()){
                            totalNum++;
                        }
                    }finally {
                        if(resultSet != null && !resultSet.isClosed()){
                            resultSet.close();
                        }
                    }
                    tableNameStr.setLength(0);
                }else{
                    tableNameStr.append(",");
                }
            }
        }catch (Exception e){
            throw e;
        }
        return Map.of("totalTime", totalTime, "totalNum", totalNum);
    }

    public Map<String,Long> executeHistoryValue(SelectTask selectTask) throws Exception{
        long totalTime = 0;
        long totalNum = 0;
        try(Connection connection = getConnection(selectTask);
            Statement statement = connection.createStatement()){
            statement.execute("USE "+selectTask.getDbName());
            String prefix = selectTask.getPrefix();
            long startIndex = selectTask.getStartIndex();
            long endIndex = selectTask.getTagNum() + startIndex;
            String startTime = selectTask.getStartTime();
            String endTime = selectTask.getEndTime();
            long startTimeStamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(startTime).getTime();
            long endTimeStamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(endTime).getTime();
            int maxSqlLength = 100000;
            StringBuilder executeSql = new StringBuilder();
            int interval = selectTask.getTimeInterval();
            String TEMP_SQL = interval > 0 ? SELECT_HISTORY_INTERVAL : SELECT_HISTORY;
            for(long i=startIndex;i<endIndex;i++){
                if(interval > 0){
                    executeSql.append(String.format(TEMP_SQL, prefix+i, startTimeStamp, endTimeStamp,interval+"m"));
                }else{
                    executeSql.append(String.format(TEMP_SQL, prefix+i, startTimeStamp, endTimeStamp));
                }
                if(executeSql.length() > maxSqlLength || i == endIndex-1){
                    long start = 0L;
                    long end = 0L;
                    ResultSet resultSet = null;
                    try {
                        synchronized (this) {
                            start = System.currentTimeMillis();
                            resultSet = statement.executeQuery(executeSql.toString());
                            end = System.currentTimeMillis();
                        }
                        totalTime = totalTime + (end - start);
                        while (resultSet.next()) {
                            totalNum++;
                        }
                    }finally {
                        if(resultSet != null && !resultSet.isClosed()){
                            resultSet.close();
                        }
                    }
                    executeSql.setLength(0);
                }else{
                    executeSql.append(" union all ");
                }
            }
        }catch (Exception e){
            throw e;
        }
        return Map.of("totalTime", totalTime, "totalNum", totalNum);
    }

    public Boolean delete(String id) {
        return selectTaskMapper.deleteById(id) > 0;
    }

    public List<SelectTask> list(SelectTask selectTask) {
        LambdaQueryWrapper<SelectTask> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(SelectTask::getCurHost,  selectTask.getCurHost());
        queryWrapper.orderByDesc(SelectTask::getCreateTime);
        return selectTaskMapper.selectList(queryWrapper);
    }
}
